package uic.edu.hk;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.util.StringTokenizer;
public class Task4 {
    public static class SearchMapper extends Mapper<Object, Text, Text, Text>{

        Text keyInfo = new Text();
        Text valueInfo = new Text();
        FileSplit split;

        public String modify(String str)
        {
            String temp = str.replaceAll("[-_\'\";:?/.,><!?`~*&^%$#@()+=|{}]", "");
            return temp;
        }
        public void map(Object key, Text value,Context context)
                throws IOException, InterruptedException {

            split = (FileSplit) context.getInputSplit();
           // System.out.println("offset"+key);
          //  System.out.println("value"+value);
            StringTokenizer itr = new StringTokenizer( value.toString());
            while( itr.hasMoreTokens() ){
                String word;
                word=modify(itr.nextToken());
                keyInfo.set(word);
                valueInfo.set(split.getPath().toString()+"："+key.toString());
                context.write(keyInfo, valueInfo);
            }
           // System.out.println("key"+keyInfo);
            //System.out.println("value"+valueInfo);
        }
    }

    /*public static class SearchCombiner extends Reducer<Text,Text,Text,Text>{
        Text info = new Text();
        public void reduce(Text key,Iterable<Text>values,Context context)throws IOException, InterruptedException{
            int sum = 0;
            for(Text value:values){
                sum+=Integer.parseInt(value.toString());
            }
            String sumAll=String.valueOf(sum);
            int index = key.toString().indexOf("+");
            info.set(key.toString().substring(index+1)+":"+sumAll);
            key.set(key.toString().substring(0,index));
            context.write(key,info);
            System.out.println("key"+key);
            System.out.println("value"+info);
        }
    }*/

    public static class SearchReducer extends Reducer<Text, Text, Text, Text>{

        private Text result = new Text();
        protected void reduce(Text key, Iterable<Text> values,Context context)
                throws IOException, InterruptedException {
            String fileList = "";
            for (Text value : values) {
                fileList += value.toString()+"+";
            }
            result.set(fileList);
            context.write(key, result);
        }
    }

    public static void main(String[] args) throws Exception {
            Configuration conf = new Configuration();

            Job job = Job.getInstance(conf,"SearchIndex");
            job.setJarByClass(Task4.class);

            job.setMapperClass(Task4.SearchMapper.class);
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(Text.class);

            Path outputPath = new Path(args[1]);
            FileSystem fileSystem =FileSystem.get(conf);
            if (fileSystem.exists(outputPath)){
                fileSystem.delete(outputPath,true);
                System.out.println("outputPath: "+args[1]+" exists, but has been deleted.");
            }

            //job.setCombinerClass(SearchCombiner.class);
            job.setReducerClass(Task4.SearchReducer.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(Text.class);

            FileInputFormat.addInputPath(job, new Path(args[0]));
            FileOutputFormat.setOutputPath(job, new Path(args[1]));
            System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}
